package com.air.antispider.stream.dataprocess.businessprocess

import com.air.antispider.stream.common.util.jedis.{JedisConnectionUtil, PropertiesUtil}
import org.apache.spark.rdd.RDD
import org.json4s.DefaultFormats
import org.json4s.jackson.Json
import redis.clients.jedis.JedisCluster

/**
  * 链路统计功能
  */
object BusinessProcess {
  def linkCount(messageRDD: RDD[String]) = {
    //数据采集量
    val serverCountMapRDD: RDD[(String, Int)] = messageRDD.map(line => {
      //将每条消息进行切割
      val arr: Array[String] = line.split("#CS#")
      //进行判断
      if (arr.length > 9) {
        var ip = arr(9)
        (ip, 1)
      } else {
        ("", 1)
      }
    })
    //进行聚合操作
    //reduceByKey((老的数据,新的数据) => 老的数据+新的数据)
    val serverCountReduce: RDD[(String, Int)] = serverCountMapRDD.reduceByKey(_ + _)
    //当前活跃连接数
    val activeCountMapRDD: RDD[(String, Int)] = messageRDD.map(line => {
      //将每条消息进行切割
      val arr: Array[String] = line.split("#CS#")
      //进行判断,返回服务器ip和活跃连接数
      if (arr.length > 11) {
        var ip = arr(9)
        var num = arr(11).toInt
        (ip, num)
      } else {
        ("", 1)
      }
    })
    //进行聚合操作，只取最新的活跃连接数即可
    val activeCountReduce: RDD[(String, Int)] = activeCountMapRDD.reduceByKey((x, y) => y)
    //将数据存储到redis中
    //先进行非空判断
    if (!serverCountReduce.isEmpty() && !activeCountReduce.isEmpty()) {
      //取出数据
      val serversCountMap: collection.Map[String, Int] = serverCountReduce.collectAsMap()
      val activeNumMap: collection.Map[String, Int] = activeCountReduce.collectAsMap()
      //创建一个Map来封装最后的数据
      var dataMap: Map[String, collection.Map[String, Int]] = Map[String, collection.Map[String, Int]]()
      dataMap += ("serversCountMap" -> serversCountMap)
      dataMap += ("activeNumMap" -> activeNumMap)
      //将map转换成json
      val json: String = Json(DefaultFormats).write(dataMap)
      //将json数据放入Redis
      val jedis: JedisCluster = JedisConnectionUtil.getJedisCluster
      val key: String = PropertiesUtil.getStringByKey("cluster.key.monitor.linkProcess", "jedisConfig.properties") + System.currentTimeMillis()
      val ex: Int = PropertiesUtil.getStringByKey("cluster.exptime.monitor", "jedisConfig.properties").toInt
      //设置数据
      //      jedis.set(key, json)
      //设置数据的有效期为2分钟
      jedis.setex(key, ex, json)
    }
  }

}
